package com.lemon.exam.common.handler;

import com.github.benmanes.caffeine.cache.Cache;
import com.github.benmanes.caffeine.cache.Caffeine;
import com.lemon.exam.common.constant.ApiConst;
import com.lemon.exam.common.constant.RedisKeyConst;
import com.lemon.exam.common.entity.dto.*;
import com.lemon.exam.common.entity.system.LoginUser;
import com.lemon.exam.common.entity.system.SysWeb;
import com.lemon.exam.common.entity.vo.AsideMenuVO;
import com.lemon.exam.common.entity.vo.TreeSelectStrVO;
import com.lemon.exam.common.entity.vo.WebRouterVO;
import com.lemon.exam.common.enums.MinioBucketEnum;
import com.lemon.exam.common.enums.SmsEnum;
import com.lemon.exam.common.exception.CustomException;
import com.lemon.exam.common.properties.ConfigProperties;
import com.lemon.exam.common.redis.RedisService;
import com.lemon.exam.common.service.FileService;
import com.lemon.exam.common.util.*;
import jakarta.annotation.Resource;
import jakarta.validation.Validator;
import lombok.extern.slf4j.Slf4j;
import org.springframework.http.codec.multipart.FilePart;
import org.springframework.r2dbc.core.DatabaseClient;
import org.springframework.stereotype.Component;
import org.springframework.transaction.reactive.TransactionalOperator;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;
import org.springframework.web.reactive.function.server.HandlerFunction;
import org.springframework.web.reactive.function.server.ServerRequest;
import org.springframework.web.reactive.function.server.ServerResponse;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;
import reactor.util.function.Tuple2;

import java.awt.*;
import java.time.Duration;
import java.util.List;
import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;
import java.util.stream.Collectors;

/**
 * 路由函数处理
 *
 * @author Lemon
 * @since 2025/4/6 21:33
 */
@Slf4j
@Component
public class RouterFunctionHandler {
    private static final int SEND_CODE_VALIDITY_MINUTES = 5;//验证码有效期 5分钟
    private static final int SEND_CODE_RATE_LIMIT_MINUTES = 30;//发送验证码的频率限制
    private static final int SEND_CODE_MAX_TIMES = 5;//发送验证码最多 5 次
    private static final String EMAIL_CONTENT = "欢迎使用Lemon考试系统，您的验证码是：%s，有效期%d分钟，请妥善保管！";
    private static final Color CAPTCHA_COLOR = new Color(1.0f, 1.0f, 1.0f, 0.1f);
    @Resource
    private RedisService redisService;
    @Resource
    private DatabaseClient databaseClient;
    @Resource
    private MailUtil mailUtil;
    @Resource
    private SmsUtil smsUtil;
    @Resource
    private FileService fileService;
    @Resource
    private NoticeHandler noticeHandler;
    @Resource
    private ConfigProperties configProperties;
    @Resource
    private Validator validator;
    @Resource
    private TransactionalOperator transactionalOperator;


    /**
     * 获取验证码图片
     *
     * @return
     */
    public HandlerFunction<ServerResponse> getCaptcha() {
        return request -> CaptchaUtil.generate(CAPTCHA_COLOR)
                .flatMap(captcha -> redisService.string().set(RedisKeyConst.LOGIN_USER_CAPTCHA + captcha.uuid(), captcha.code(), Duration.ofMinutes(1))
                        .then(CaptchaUtil.createBase64Response(captcha, "PNG")))
                .onErrorResume(ResponseUtil.err());
    }

    /**
     * 用户注册
     *
     * @return
     */
    public HandlerFunction<ServerResponse> register() {
        return request -> request.bodyToMono(RegisterUserDTO.class)
                .flatMap(user -> ValidatorUtil.validate(validator, user))
                .flatMap(this::validateAndCheck)
                .flatMap(this::insertUser)
                .onErrorResume(ResponseUtil.err());
    }

    /**
     * 验证用户信息并检查输入数据是否正确
     *
     * @param user
     * @return
     */
    private Mono<RegisterUserDTO> validateAndCheck(RegisterUserDTO user) {
        return Mono.zip(
                this.count("username", user.username(), "用户名"),
                this.count("email", user.email(), "邮箱"),
                this.count("phone", user.phone(), "手机号")
        ).thenReturn(user);
    }

    /**
     * 检查用户输入的数据是否正确
     *
     * @param field
     * @param value
     * @param fieldName
     * @return
     */
    private Mono<Boolean> count(String field, String value, String fieldName) {
        return databaseClient.sql("SELECT 1 FROM sys_user WHERE %s = :value LIMIT 1".formatted(field))
                .bind("value", value)
                .map((row, metadata) -> true)
                .first()
                .hasElement()
                .flatMap(exists -> exists ? Mono.error(new CustomException(fieldName + "已存在")) : Mono.just(true))
                .subscribeOn(Schedulers.boundedElastic());
    }

    /**
     * 密码解密并执行插入用户
     *
     * @param user
     * @return
     */
    private Mono<ServerResponse> insertUser(RegisterUserDTO user) {
        return Mono.fromCallable(() -> RsaUtil.getInstance().decryptByPrivateKey(configProperties.getRsa().getPrivateKey(), user.password()))
                .subscribeOn(Schedulers.boundedElastic())
                .map(SecurityUtil::passwordEncode)
                .flatMap(encryptedPassword -> insertUser(user, encryptedPassword))
                .flatMap(ResponseUtil.ok())
                .onErrorResume(ResponseUtil.err());
    }

    private static final String INSERT_USER_SQL = """
            INSERT INTO sys_user
            (username, password, name, phone, sex, email, birthday, province_id, city_id, county_id, address, create_by, create_time, status, deleted) VALUES
            (:username, :password, :name, :phone, :sex, :email, :birthday, :provinceId, :cityId, :countyId, :address, :name, NOW(), 1, 0)
            """;

    /**
     * 插入用户
     *
     * @param user
     * @param password
     * @return
     */
    private Mono<String> insertUser(RegisterUserDTO user, String password) {
        return databaseClient.sql(INSERT_USER_SQL)
                .bind("username", user.username())
                .bind("password", password)
                .bind("name", user.name())
                .bind("phone", user.phone())
                .bind("sex", user.sex())
                .bind("email", user.email())
                .bind("birthday", user.birthday())
                .bind("provinceId", user.provinceId())
                .bind("cityId", user.cityId())
                .bind("countyId", user.countyId())
                .bind("address", user.address())
                .fetch()
                .rowsUpdated()
                .flatMap(updated -> updated == 1 ? Mono.just("注册成功") : Mono.error(new CustomException("注册失败")));
    }

    private static final String GET_CURRENT_USER_SQL = """
            SELECT
                id,
                username,
                name,
                phone,
                sex,
                email,
                birthday,
                avatar,
                province_id provinceId,
                city_id cityId,
                county_id countyId,
                address,
                SUBSTRING_INDEX(prev_login_ip, ',', 1) prevLoginIp,
                SUBSTRING_INDEX(prev_login_time, ',', 1) prevLoginTime,
                create_by createBy,
                create_time createTime,
                update_by updateBy,
                update_time updateTime,
                status
            FROM
                sys_user
            WHERE
                id = :id
                AND deleted = 0
                LIMIT 1
            """;

    /**
     * 根据ID获取用户信息
     *
     * @param userId
     * @return
     */
    private Mono<Map<String, Object>> getUserById(Long userId) {
        return databaseClient.sql(GET_CURRENT_USER_SQL).bind("id", userId)
                .fetch()
                .one()
                .subscribeOn(Schedulers.boundedElastic());
    }

    private static final String GET_ROLE_NAME_LIST_SQL = """
            SELECT DISTINCT name
            FROM
                sys_role
            WHERE
                status = 1
                AND deleted = 0
                AND id IN (SELECT role_id FROM sys_user_role WHERE user_id = :userId)
                AND name IS NOT NULL
            """;

    /**
     * 根据ID获取角色名称
     *
     * @param userId
     * @return
     */
    private Flux<String> getRoleNameListByUserId(Long userId) {
        return databaseClient.sql(GET_ROLE_NAME_LIST_SQL)
                .bind("userId", userId)
                .map(row -> row.get("name", String.class))
                .all()
                .subscribeOn(Schedulers.boundedElastic());
    }

    /**
     * 获取当前用户信息
     *
     * @return
     */
    public HandlerFunction<ServerResponse> getCurrentUser() {
        return request -> SecurityUtil.getUserId()
                .flatMap(userId -> Mono.zip(this.getUserById(userId), this.getRoleNameListByUserId(userId).collectList())
                        .map(TO_USER_MAP)
                        .flatMap(ResponseUtil.ok())
                        .onErrorResume(ResponseUtil.err()));
    }

    /**
     * 合并当前用户信息
     */
    private static final Function<Tuple2<Map<String, Object>, List<String>>, Map<String, Object>> TO_USER_MAP = tuple -> {
        Map<String, Object> user = tuple.getT1();
        Object provinceId = user.get("provinceId");
        Object cityId = user.get("cityId");
        Object countyId = user.get("countyId");

        //处理城市信息
        user.put("provinceId", provinceId == null ? null : Integer.parseInt(provinceId.toString()));
        user.put("cityId", cityId == null ? null : Integer.parseInt(cityId.toString()));
        user.put("countyId", countyId == null ? null : Integer.parseInt(countyId.toString()));
        user.put("roleNameList", tuple.getT2());
        return user;
    };

    /**
     * 获取当前用户的头像地址
     *
     * @return
     */
    public HandlerFunction<ServerResponse> getCurrentUserAvatar() {
        return request -> SecurityUtil.getUserId()
                .flatMap(userId -> databaseClient.sql("SELECT COALESCE(avatar, '') as avatar FROM sys_user WHERE id = :id LIMIT 1")
                        .bind("id", userId)
                        .map(row -> row.get("avatar", String.class))
                        .one()
                        .flatMap(ResponseUtil.ok()));
    }

    private static final String GET_MENU_PATH_SQL = """
            SELECT DISTINCT menu_path
            FROM
            	view_role_menu_resource
            WHERE
            	role_status = 1
            	AND role_deleted = 0
            	AND menu_status = 1
            	AND menu_path IS NOT NULL
            	AND role_id IN (SELECT role_id FROM sys_user_role WHERE user_id = :userId)
            """;

    private Flux<String> getMenuPathByUserId(Long userId) {
        return databaseClient.sql(GET_MENU_PATH_SQL)
                .bind("userId", userId)
                .map(row -> row.get("menu_path", String.class))
                .all()
                .subscribeOn(Schedulers.boundedElastic());
    }

    /**
     * 获取路由信息
     *
     * @return
     */
    public HandlerFunction<ServerResponse> getRouters() {
        return request -> SecurityUtil.getUserId().flatMap(userId -> Mono.zip(
                        this.getMenuPathByUserId(userId).collectList(),
                        redisService.set().get(RedisKeyConst.SYSTEM_INTERFACE, SysWeb.class).collectList().subscribeOn(Schedulers.boundedElastic())
                ).map(tuple -> tuple.getT2()
                        .parallelStream()
                        .filter(web -> tuple.getT1().contains(web.path()))
                        .map(web -> new WebRouterVO(
                                web.parentUri(),
                                web.parentComponent(),
                                web.childUri(),
                                web.childComponent()
                        ))
                        .toList())
                .flatMap(ResponseUtil.ok()));
    }

    /**
     * 获取路由和资源信息
     *
     * @return
     */
    public HandlerFunction<ServerResponse> getRouterResource() {
        return request -> SecurityUtil.getUserId().flatMap(userId -> Mono.zip(
                        this.getMenuPathResourcePathByUserId(userId).collectList(),
                        redisService.set().get(RedisKeyConst.SYSTEM_INTERFACE, SysWeb.class).collectList().subscribeOn(Schedulers.boundedElastic())
                ))
                .map(TO_RESOURCE_MAP)
                .flatMap(ResponseUtil.ok())
                .onErrorResume(ResponseUtil.err(500));
    }

    private static final String GET_MENU_PATH_RESOURCE_PATH_SQL = """
            SELECT
                CONCAT(menu_path, resource_path) AS path
            FROM
                view_role_menu_resource
            WHERE
                role_status = 1
                AND role_deleted = 0
                AND menu_status = 1
                AND menu_path IS NOT NULL
                AND resource_path IS NOT NULL
                AND role_id IN (SELECT role_id FROM sys_user_role WHERE user_id = :userId)
            """;

    /**
     * 根据用户ID获取菜单路径和资源路径
     *
     * @param userId
     * @return
     */
    private Flux<String> getMenuPathResourcePathByUserId(Long userId) {
        return databaseClient.sql(GET_MENU_PATH_RESOURCE_PATH_SQL)
                .bind("userId", userId)
                .map(row -> row.get("path", String.class))
                .all()
                .subscribeOn(Schedulers.boundedElastic());
    }

    /**
     * 合并资源信息
     */
    private static final Function<Tuple2<List<String>, List<SysWeb>>, Map<String, List<String>>> TO_RESOURCE_MAP = tuple -> {
        //1.资源路径
        List<String> pathList = tuple.getT1();
        if (CollectionUtils.isEmpty(pathList)) {
            return Collections.emptyMap();
        }
        return tuple.getT2()
                .stream()
                .flatMap(web -> web.methods()
                        .stream()
                        .filter(method -> pathList.contains(method.path()))
                        .map(method -> Map.entry(web.parentUri() + "/" + web.childUri(), method.sign()))
                ).collect(Collectors.groupingBy(
                        Map.Entry::getKey,
                        Collectors.mapping(Map.Entry::getValue, Collectors.toList())
                ));
    };

    /**
     * 修改密码
     *
     * @return
     */
    public HandlerFunction<ServerResponse> updatePasswordByEmailOrPhone() {
        return request -> request.bodyToMono(UpdatePasswordDTO.class)
                .flatMap(params -> ValidatorUtil.validate(validator, params, "email".equals(params.active()) ? "phone" : "email"))
                .flatMap(this::verifyCode)
                .flatMap(this::processPasswordUpdate)
                .flatMap(ResponseUtil.ok())
                .onErrorResume(ResponseUtil.err());
    }

    /**
     * 判断验证码是否正确
     *
     * @param param
     * @return
     */
    private Mono<UpdatePasswordDTO> verifyCode(UpdatePasswordDTO param) {
        String key = RedisKeyConst.UPDATE_PASSWORD_CODE + ("email".equals(param.active()) ? param.email() : param.phone());

        return redisService.string().get(key, String.class)
                .switchIfEmpty(Mono.error(new CustomException("验证码已失效")))
                .filter(code -> code.equals(param.code()))
                .switchIfEmpty(Mono.error(new CustomException("验证码错误")))
                .thenReturn(param)
                .publishOn(Schedulers.boundedElastic())
                .doOnSuccess(v -> redisService.string().del(key).subscribe());
    }

    /**
     * 处理密码更新
     *
     * @param param
     * @return
     */
    private Mono<String> processPasswordUpdate(UpdatePasswordDTO param) {
        return Mono.fromCallable(() -> {
                    String privateKey = configProperties.getRsa().getPrivateKey();
                    RsaUtil rsaUtil = RsaUtil.getInstance();
                    String decryptNewPassword = rsaUtil.decryptByPrivateKey(privateKey, param.newPassword());
                    String decryptConfirmPassword = rsaUtil.decryptByPrivateKey(privateKey, param.confirmPassword());
                    if (!decryptNewPassword.equals(decryptConfirmPassword)) {
                        throw new CustomException("两次输入的密码不一致");
                    }
                    //密码加密
                    return SecurityUtil.passwordEncode(decryptNewPassword);
                })
                .subscribeOn(Schedulers.boundedElastic()) // 密码处理在弹性线程池
                .flatMap(encodedPassword -> updatePassword(param, encodedPassword));
    }

    private static final String UPDATE_PASSWORD_SQL = """
            UPDATE sys_user SET password = :password WHERE
            (CASE
            WHEN :active = 'email' THEN email = :email
            WHEN :active = 'phone' THEN phone = :phone
            ELSE FALSE
            END)
            """;

    /**
     * 执行修改密码
     *
     * @param param
     * @param password
     * @return
     */
    private Mono<String> updatePassword(UpdatePasswordDTO param, String password) {
        return databaseClient.sql(UPDATE_PASSWORD_SQL)
                .bind("password", password)
                .bind("active", param.active())
                .bind("email", param.email())
                .bind("phone", param.phone())
                .fetch()
                .rowsUpdated()
                .flatMap(rowsUpdated -> rowsUpdated == 1 ? Mono.just("密码修改成功") : Mono.error(new CustomException("用户不存在或更新失败")));

    }

    /**
     * 发送验证码
     *
     * @return
     */
    public HandlerFunction<ServerResponse> sendCodeByEmailOrPhone() {
        return request -> request.bodyToMono(UpdatePasswordDTO.class)
                .flatMap(param -> "email".equals(param.active()) ? validateAndProcessEmail(param.email()) : validateAndProcessPhone(param.phone()))
                .map(data -> Map.of("data", data, "time", SEND_CODE_VALIDITY_MINUTES))
                .flatMap(ResponseUtil.ok())
                .onErrorResume(ResponseUtil.err());
    }

    /**
     * 验证邮箱并生成验证码
     *
     * @param email
     * @return
     */
    private Mono<String> validateAndProcessEmail(String email) {
        return ValidatorUtil.validateProperty(validator, email, "email")
                .then(checkUserExists("email", email))
                .then(checkRateLimit(email))
                .then(generateAndSendEmailCode(email, RandomUtil.num(6)));
    }

    /**
     * 验证手机号并生成验证码
     *
     * @param phone
     * @return
     */
    private Mono<String> validateAndProcessPhone(String phone) {
        return ValidatorUtil.validateProperty(validator, phone, "phone")
                .then(checkUserExists("phone", phone))
                .then(checkRateLimit(phone))
                .then(generateAndSendSmsCode(phone, RandomUtil.num(6)));
    }

    /**
     * 检查用户是否存在
     *
     * @param field
     * @param value
     * @return
     */
    private Mono<Void> checkUserExists(String field, String value) {
        return databaseClient.sql("SELECT 1 FROM sys_user WHERE %s = :value AND status = 1 AND deleted = 0 LIMIT 1".formatted(field))
                .bind("value", value)
                .map((row, metadata) -> true)
                .one()
                .then()
                .onErrorResume(NoSuchElementException.class, e -> Mono.error(new CustomException("email".equals(field) ? "邮箱账号不存在" : "手机号不存在"))); // 没有数据就报错
    }

    /**
     * 检查用户操作频率
     *
     * @param identifier
     * @return
     */
    private Mono<Void> checkRateLimit(String identifier) {
        return Mono.defer(() -> {
            final String key = RedisKeyConst.UPDATE_PASSWORD_COUNT + identifier;
            return redisService.string().get(key, Integer.class).defaultIfEmpty(0)// 先检查是否已存在，不存在设置为0
                    .flatMap(count -> {
                        if (count >= SEND_CODE_MAX_TIMES) {
                            return Mono.error(new CustomException("操作频繁，请稍后再试"));
                        }
                        return redisService.string().increment(key, 1).flatMap(newCount -> {// 增加计数
                            if (newCount.equals(1L)) {// 第一次设置，添加过期时间
                                return redisService.string().expire(key, Duration.ofMinutes(SEND_CODE_RATE_LIMIT_MINUTES)).thenReturn(newCount);
                            }
                            return Mono.just(newCount);
                        });
                    }).then();
        });
    }

    /**
     * 发送邮箱验证码
     *
     * @param email
     * @return
     */
    private Mono<String> generateAndSendEmailCode(String email, String code) {
        return redisService.string().set(RedisKeyConst.UPDATE_PASSWORD_CODE + email, code, Duration.ofMinutes(SEND_CODE_VALIDITY_MINUTES))
                .then(mailUtil.send(email, "验证码", EMAIL_CONTENT.formatted(code, SEND_CODE_VALIDITY_MINUTES), false))
                .thenReturn("邮件发送成功！");
    }

    /**
     * 发送短信验证码
     *
     * @param phone
     * @return
     */
    private Mono<String> generateAndSendSmsCode(String phone, String code) {
        return redisService.string().set(RedisKeyConst.UPDATE_PASSWORD_CODE + phone, code, Duration.ofMinutes(SEND_CODE_VALIDITY_MINUTES))
                .then(smsUtil.send(phone, Map.of("code", code), SmsEnum.FORGOT_PASSWORD))
                .flatMap(ok -> ok ? Mono.just("短信发送成功！") : Mono.error(new CustomException("短信发送失败")));
    }

    /**
     * 修改当前用户信息
     *
     * @return
     */
    public HandlerFunction<ServerResponse> updateCurrentUser() {
        return request -> Mono.zip(request.bodyToMono(UpdateUserDTO.class), SecurityUtil.getUserId())
                .flatMap(tuple -> {
                    UpdateUserDTO param = tuple.getT1();
                    Long userId = tuple.getT2();
                    if (!param.id().equals(userId)) {
                        return ResponseUtil.err("无权限操作！");
                    } else {
                        return Mono.zip(
                                this.count("email", param.email(), "邮箱", userId),
                                this.count("phone", param.phone(), "手机号", userId)
                        ).then(updateCurrentUser(param));
                    }
                })
                .onErrorResume(ResponseUtil.err());
    }

    /**
     * 查询个数
     *
     * @param field
     * @param value
     * @param fieldName
     * @param userId
     * @return
     */
    private Mono<Integer> count(String field, String value, String fieldName, Long userId) {
        return databaseClient.sql("SELECT COUNT(1) AS num FROM sys_user WHERE %s = :value AND id != :userId".formatted(field))
                .bind("value", value)
                .bind("userId", userId)
                .map(row -> row.get("num", Integer.class))
                .one()
                .filter(count -> count == 0)
                .switchIfEmpty(Mono.error(new CustomException(fieldName + "已存在！")))
                .defaultIfEmpty(0)
                .subscribeOn(Schedulers.parallel());
    }

    private static final String UPDATE_CURRENT_USER_SQL = """
            UPDATE sys_user
            SET name = :name,
            email = :email,
            phone = :phone,
            sex = :sex,
            birthday = :birthday,
            avatar = :avatar,
            province_id = :provinceId,
            city_id = :cityId,
            county_id = :countyId,
            address = :address,
            update_by = :name
            WHERE id = :id
            """;

    /**
     * 修改当前用户信息
     *
     * @param param
     * @return
     */
    private Mono<ServerResponse> updateCurrentUser(UpdateUserDTO param) {
        return databaseClient.sql(UPDATE_CURRENT_USER_SQL)
                .bindProperties(param)
                .fetch()
                .rowsUpdated()
                .flatMap(count -> count == 1 ? ResponseUtil.ok("修改成功！") : ResponseUtil.err(500, "修改失败"));
    }

    /**
     * 修改当前用户密码
     *
     * @return
     */
    public HandlerFunction<ServerResponse> updateCurrentUserPassword() {
        return request -> request.bodyToMono(ChangePasswordDTO.class)
                .flatMap(this::comparePassword)
                .flatMap(param -> ValidatorUtil.validate(validator, param))
                .flatMap(param -> SecurityUtil.getUserId()
                        .flatMap(userId -> getPasswordAndCompare(userId, param.oldPassword())
                                .then(updatePassword(userId, SecurityUtil.passwordEncode(param.newPassword())))));
    }

    /**
     * 获取当前用户密码并比较
     *
     * @param userId
     * @param oldPassword
     * @return
     */
    private Mono<String> getPasswordAndCompare(Long userId, String oldPassword) {
        return databaseClient.sql("SELECT password FROM sys_user WHERE id = :id LIMIT 1")
                .bind("id", userId)
                .map(row -> row.get("password", String.class))
                .one()
                .filter(password -> SecurityUtil.passwordMatch(oldPassword, password))
                .switchIfEmpty(Mono.error(new CustomException("旧密码错误！")));
    }

    /**
     * 比较新密码和确认密码是否一致
     *
     * @param param
     * @return
     */
    private Mono<ChangePasswordDTO> comparePassword(ChangePasswordDTO param) {
        return Mono.fromCallable(() -> {
                    RsaUtil rsaUtil = RsaUtil.getInstance();
                    final String privateKey = configProperties.getRsa().getPrivateKey();
                    final String decryptOldPassword = rsaUtil.decryptByPrivateKey(privateKey, param.oldPassword());
                    final String decryptNewPassword = rsaUtil.decryptByPrivateKey(privateKey, param.newPassword());
                    final String decryptConfirmPassword = rsaUtil.decryptByPrivateKey(privateKey, param.confirmPassword());
                    return new ChangePasswordDTO(decryptOldPassword, decryptNewPassword, decryptConfirmPassword);
                }).filter(dto -> dto.newPassword().equals(dto.confirmPassword()))
                .switchIfEmpty(Mono.error(new CustomException("新密码与确认密码不一致！")))
                .defaultIfEmpty(param);
    }

    /**
     * 修改密码
     *
     * @param userId
     * @param password
     * @return
     */
    private Mono<ServerResponse> updatePassword(Long userId, String password) {
        return databaseClient.sql("UPDATE sys_user SET password = :password WHERE id = :id LIMIT 1")
                .bind("password", password)
                .bind("id", userId)
                .fetch()
                .rowsUpdated()
                .flatMap(updated -> updated == 1 ? ResponseUtil.ok("修改成功，即将返回登录页...") : ResponseUtil.err(500, "修改失败！"));
    }

    /**
     * 上传当前用户头像
     *
     * @return
     */
    public HandlerFunction<ServerResponse> uploadCurrentUserAvatar() {
        return request -> request.multipartData()
                //获取文件部分（前端上传的字段名是 "file"）
                //.flatMap(parts -> fileService.uploadToLocal((FilePart) parts.toSingleValueMap().get("file"), "avatar").map(file -> Map.of("url", file.url() + file.path())))//上传到本地
                .flatMap(parts -> fileService.uploadToMinio((FilePart) parts.toSingleValueMap().get("file"), MinioBucketEnum.AVATAR).map(file -> Map.of("url", file.url())))//上传到MinIO
                .flatMap(ResponseUtil.ok());
    }

    /**
     * 获取当前用户通知数量-SSE
     *
     * @return
     */
    public HandlerFunction<ServerResponse> getNoticeCountConnection() {
        return noticeHandler.connectionHandler();
    }

    /**
     * 菜单缓存
     */
    private final Cache<Long, List<AsideMenuVO>> menuCache = Caffeine.newBuilder()
            .expireAfterWrite(10, TimeUnit.MINUTES)
            .maximumSize(10000)
            .build();

    /**
     * 获取当前用户菜单 (先从缓存中获取，缓存没有查数据库和Redis)
     *
     * @return
     */
    public HandlerFunction<ServerResponse> getAsideMenu() {
        return request -> SecurityUtil.getUserId()
                .flatMap(this::getCachedMenu)
                .flatMap(ResponseUtil.ok())
                .onErrorResume(ResponseUtil.err());
    }

    /**
     * 从缓存中获取菜单数据
     *
     * @param userId
     * @return
     */
    private Mono<List<AsideMenuVO>> getCachedMenu(Long userId) {
        return Mono.fromCallable(() -> menuCache.get(userId, this::loadMenuData)).subscribeOn(Schedulers.boundedElastic());
    }

    /**
     * 从缓存中获取菜单数据
     *
     * @param userId
     * @return
     */
    private List<AsideMenuVO> loadMenuData(Long userId) {
        return Mono.zip(
                        this.getAsideMenu(userId).collectList(),
                        redisService.set().get(RedisKeyConst.SYSTEM_INTERFACE, SysWeb.class).collectList().subscribeOn(Schedulers.boundedElastic())
                )
                .flatMap(tuple -> this.processMenuData(tuple.getT1(), tuple.getT2()))
                .blockOptional(Duration.ofSeconds(10)).orElseGet(Collections::emptyList);
    }

    private static final String GET_ASIDE_MENU_SQL = """
            SELECT
                id, name, path, parent_id, icon, seq, tree_link_id
            FROM sys_menu
            WHERE
                status = 1
                AND id IN (SELECT menu_id FROM sys_role_menu_resource WHERE role_id IN (SELECT role_id FROM sys_user_role WHERE user_id = :userId)) ORDER BY seq
            """;

    /**
     * 获取菜单数据
     *
     * @param userId
     * @return
     */
    private Flux<AsideMenuVO> getAsideMenu(Long userId) {
        return databaseClient.sql(GET_ASIDE_MENU_SQL)
                .bind("userId", userId)
                .map(row -> {
                    AsideMenuVO menu = new AsideMenuVO();
                    menu.setId(row.get("id", Long.class));
                    menu.setName(row.get("name", String.class));
                    menu.setPath(row.get("path", String.class));
                    menu.setParentId(row.get("parent_id", Long.class));
                    menu.setIcon(row.get("icon", String.class));
                    menu.setSeq(row.get("seq", Integer.class));
                    menu.setTreeLinkId(row.get("tree_link_id", String.class));
                    return menu;
                })
                .all()
                .subscribeOn(Schedulers.boundedElastic());
    }

    /**
     * 处理菜单数据
     *
     * @param menuList
     * @param webList
     * @return
     */
    private Mono<List<AsideMenuVO>> processMenuData(List<AsideMenuVO> menuList, List<SysWeb> webList) {
        if (CollectionUtils.isEmpty(menuList)) {
            return Mono.just(Collections.emptyList());
        }

        return Mono.fromCallable(() -> {
            //1.创建web路径映射
            Map<String, SysWeb> webMap = webList.stream().collect(Collectors.toMap(SysWeb::path, Function.identity()));

            //2.处理菜单基本属性
            menuList.forEach(menu -> {
                //设置名称和URI
                SysWeb web = webMap.get(menu.getPath());
                if (web != null) {
                    if (!StringUtils.hasText(menu.getName())) {
                        menu.setName(web.name());
                    }
                    menu.setUri(web.parentUri() + "/" + web.childUri());
                }
            });

            //3.创建菜单ID到名称的映射
            Map<Long, String> menuIdNameMap = menuList.stream().collect(Collectors.toMap(AsideMenuVO::getId, AsideMenuVO::getName));

            //4.处理面包屑
            menuList.forEach(menu -> {
                if (StringUtils.hasText(menu.getTreeLinkId())) {
                    String[] idArr = menu.getTreeLinkId().split(",");
                    menu.setBread(Arrays.stream(idArr, 1, idArr.length).map(id -> menuIdNameMap.get(Long.parseLong(id))).filter(Objects::nonNull).toArray(String[]::new));
                }
            });

            //5.构建菜单树
            Map<Long, List<AsideMenuVO>> menuGroupByParent = menuList.stream().collect(Collectors.groupingBy(AsideMenuVO::getParentId));

            //6.设置子菜单并收集所有子菜单ID
            Set<AsideMenuVO> childMenus = new HashSet<>();
            menuList.forEach(menu -> {
                List<AsideMenuVO> children = menuGroupByParent.getOrDefault(menu.getId(), Collections.emptyList());
                menu.setChildren(children);
                childMenus.addAll(children);
            });

            //7.返回顶级菜单
            return menuList.stream()
                    .filter(menu -> !childMenus.contains(menu))
                    .filter(menu -> ApiConst.SUPER_PARENT_ID.equals(menu.getParentId()))
                    .sorted(Comparator.comparing(AsideMenuVO::getSeq))
                    .toList();
        });
    }


    private static final String GET_NOTICE_SQL = """
            SELECT id, title, create_time createTime FROM sys_notice WHERE
            (CASE
            WHEN :type = 'mine' THEN create_id = :userId
            WHEN :type = 'unread' THEN create_id <> :userId AND id IN (SELECT notice_id FROM sys_notice_user WHERE user_id = :userId AND is_read = 0)
            WHEN :type = 'read' THEN create_id <> :userId AND id IN (SELECT notice_id FROM sys_notice_user WHERE user_id = :userId AND is_read = 1)
            ELSE FALSE
            END)
            ORDER BY create_time DESC
            """;

    /**
     * 获取通知列表
     *
     * @return
     */
    public HandlerFunction<ServerResponse> getNotice() {
        return request -> Mono.zip(SecurityUtil.getUserId(), queryParam(request, "type"))
                .flatMap(tuple -> databaseClient.sql(GET_NOTICE_SQL)
                        .bind("userId", tuple.getT1())
                        .bind("type", tuple.getT2())
                        .fetch()
                        .all()
                        .collectList()
                        .flatMap(ResponseUtil.ok()));
    }

    /**
     * 查询参数
     *
     * @param request
     * @param param
     * @return
     */
    private Mono<String> queryParam(ServerRequest request, String param) {
        return Mono.justOrEmpty(request.queryParam(param)).switchIfEmpty(Mono.error(new CustomException("缺少param参数")));
    }

    /**
     * 获取通知详情
     *
     * @return
     */
    public HandlerFunction<ServerResponse> getNoticeById() {
        return request -> {
            //1. 获取并验证参数
            Mono<String> noticeId = queryParam(request, "id");
            Mono<String> type = queryParam(request, "type");

            //2. 如果需要标记为已读，并且需要更新数据库，还要通过SSE推送数据
            return Mono.zip(SecurityUtil.getUserId(), noticeId, type)
                    .filter(tuple -> "unread".equals(tuple.getT3())) // 只有当isUnread=true时才处理
                    .flatMap(tuple -> updateNoticeUser(tuple.getT1(), tuple.getT2())
                            .flatMap(updated -> noticeHandler.pushNotice(tuple.getT1())))
                    .then(getNoticeById(noticeId, type)).flatMap(ResponseUtil.ok())
                    .onErrorResume(ResponseUtil.err(500));
        };
    }

    private static final String GET_NOTICE_BY_ID_SQL = """
            SELECT
                id,
                title,
                content,
                type,
                remark,
                create_by AS createBy,
                create_time AS createTime
            FROM sys_notice
            WHERE
                id = :id
            LIMIT 1
            """;

    /**
     * 获取通知信息
     *
     * @param noticeId
     * @return
     */
    private Mono<Map<String, Object>> getNoticeById(Mono<String> noticeId, Mono<String> type) {
        return Mono.zip(noticeId, type).flatMap(tuple -> databaseClient.sql(GET_NOTICE_BY_ID_SQL)
                .bind("id", tuple.getT1())
                .fetch()
                .one()
                .flatMap(notice -> {
                    if ("mine".equals(tuple.getT2())) {
                        return getUserNameByNotice(tuple, notice);
                    }
                    return Mono.just(notice);
                }));
    }

    /**
     * 获取通知的用户信息
     *
     * @param tuple
     * @param notice
     * @return
     */
    private Mono<Map<String, Object>> getUserNameByNotice(Tuple2<String, String> tuple, Map<String, Object> notice) {
        return databaseClient.sql("SELECT name FROM sys_user WHERE id IN (SELECT user_id FROM sys_notice_user WHERE notice_id = :noticeId)")
                .bind("noticeId", tuple.getT1())
                .map(row -> row.get("name", String.class))
                .all()
                .collectList()
                .map(nameList -> {
                    notice.put("createBy", "");
                    notice.put("userList", nameList);
                    return notice;
                });
    }

    /**
     * 更新通知用户
     *
     * @param userId
     * @param noticeId
     * @return
     */
    private Mono<Long> updateNoticeUser(Long userId, String noticeId) {
        return databaseClient.sql("UPDATE sys_notice_user SET is_read = 1 WHERE user_id = :userId AND notice_id = :noticeId")
                .bind("userId", userId)
                .bind("noticeId", noticeId)
                .fetch()
                .rowsUpdated();
    }

    /**
     * 获取班级用户树结构
     *
     * @return
     */
    public HandlerFunction<ServerResponse> getTreeClassUserByCurrent() {
        return request -> SecurityUtil.getUserId()
                .flatMap(this::fetchClassUsers)
                .flatMap(this::buildTreeStructure)
                .flatMap(ResponseUtil.ok());
    }

    private static final String GET_TREE_CLASS_USER_BY_CURRENT_SQL = """
            SELECT
                class_id, class_name, grade_id, grade_name, user_id, user_name, job_name
            FROM
                view_class_grade_user_job
            WHERE
                class_deleted = 0
                AND grade_deleted = 0
                AND user_status = 1
                AND user_deleted = 0
                AND user_id <> :userId
                AND class_id IN (SELECT class_id FROM biz_class_user WHERE user_id = :userId)
            """;

    /**
     * 获取班级用户
     *
     * @param userId
     * @return
     */
    private Mono<List<ClassUserDTO>> fetchClassUsers(Long userId) {
        return databaseClient.sql(GET_TREE_CLASS_USER_BY_CURRENT_SQL)
                .bind("userId", userId)
                .map(row -> new ClassUserDTO(
                        row.get("class_id", Long.class),
                        row.get("class_name", String.class),
                        row.get("grade_id", Long.class),
                        row.get("grade_name", String.class),
                        row.get("user_id", Long.class),
                        row.get("user_name", String.class),
                        row.get("job_name", String.class)
                ))
                .all()
                .collectList();
    }

    /**
     * 构建树形结构
     *
     * @param classUsers
     * @return
     */
    private Mono<List<TreeSelectStrVO>> buildTreeStructure(List<ClassUserDTO> classUsers) {
        return Mono.fromCallable(() -> Optional.ofNullable(classUsers).orElseGet(ArrayList::new).stream()
                        .collect(Collectors.groupingBy(ClassUserDTO::gradeId))
                        .entrySet().stream()
                        .map(grade -> new TreeSelectStrVO(
                                String.valueOf(grade.getKey()),
                                grade.getValue().get(0).gradeName(),
                                grade.getValue().stream()
                                        .collect(Collectors.groupingBy(ClassUserDTO::classId))
                                        .entrySet().stream()
                                        .map(getTreeSelectStr(grade.getKey()))
                                        .filter(Objects::nonNull)
                                        .toList()))
                        .toList())
                .subscribeOn(Schedulers.parallel());
    }

    /**
     * 获取树形结构
     *
     * @param gradeId
     * @return
     */
    private static Function<Map.Entry<Long, List<ClassUserDTO>>, TreeSelectStrVO> getTreeSelectStr(Long gradeId) {
        return clazz -> {
            var users = clazz.getValue().stream()
                    .filter(u -> u.userId() != null)
                    .map(u -> new TreeSelectStrVO(
                            String.format("%s-%s-%s", gradeId, clazz.getKey(), u.userId()),
                            u.userName() + (StringUtils.hasText(u.jobName()) ? " (" + u.jobName() + ")" : ""),
                            null)
                    ).toList();
            return users.isEmpty() ? null : new TreeSelectStrVO(
                    String.format("%s-%s", gradeId, clazz.getKey()),
                    clazz.getValue().get(0).className(),
                    users
            );
        };
    }

    /**
     * 保存或发送通知
     *
     * @return
     */
    public HandlerFunction<ServerResponse> sendNotice() {
        return request -> request.bodyToMono(NoticeDTO.class)
                .flatMap(notice -> ValidatorUtil.validate(validator, notice))
                .flatMap(notice -> SecurityUtil.getLoginUser()
                        .switchIfEmpty(Mono.error(new CustomException("用户未登录")))
                        .flatMap(loginUser -> validateAndInsertNotice(notice, loginUser))
                )
                .flatMap(ResponseUtil.ok())
                .onErrorResume(ResponseUtil.err());
    }

    /**
     * 插入通知
     *
     * @param notice
     * @param loginUser
     * @return
     */
    private Mono<String> validateAndInsertNotice(NoticeDTO notice, LoginUser loginUser) {
        return checkExamination(loginUser.getId()).flatMap(isExamining -> {
            if (isExamining) {
                return Mono.error(new CustomException("用户正在考试中，无法发送通知"));
            }
            //是否是保存通知，若是则存入redis，保存7天
            if (notice.isSave()) {
                return redisService.string().set(RedisKeyConst.SEND_NOTICE_USER_ID + loginUser.getId(), notice, Duration.ofDays(7))
                        .thenReturn("保存成功！有效期7天");
            }
            return insertNotice(notice, loginUser)
                    .flatMap(userIdList -> noticeHandler.broadcastNotice(userIdList))
                    .thenReturn("发布成功！");
        });
    }

    private static final String CHECK_EXAMINATION_SQL = """
             SELECT 1
             FROM biz_paper bp
             INNER JOIN biz_class_user bcu ON bp.class_id = bcu.class_id
             WHERE
                 bp.start_time <= NOW()
                 AND bp.end_time >= NOW()
                 AND bcu.user_id = :userId
             LIMIT 1
            """;

    /**
     * 检查用户是否正在考试
     *
     * @param userId
     * @return
     */
    private Mono<Boolean> checkExamination(Long userId) {
        return databaseClient.sql(CHECK_EXAMINATION_SQL)
                .bind("userId", userId)
                .map((row, metadata) -> true)
                .first()
                .hasElement();
    }

    /**
     * 获取用户id列表
     *
     * @param notice
     * @return
     */
    private static Mono<List<Long>> getUserIdList(NoticeDTO notice) {
        return Mono.fromCallable(() -> notice.userIdList().parallelStream().map(userId -> {
            String id = userId.split("-")[2];
            if (StringUtils.hasText(id) && !"null".equals(id)) {
                return Long.parseLong(id);
            }
            return null;
        }).filter(Objects::nonNull).toList());
    }

    /**
     * 插入通知并返回状态
     *
     * @param notice
     * @return
     */
    private Mono<List<Long>> insertNotice(NoticeDTO notice, LoginUser loginUser) {
        return getUserIdList(notice).flatMap(userIdList -> insertNoticeAndGetId(notice, loginUser)
                .flatMap(noticeId -> noticeHandler.batchInsertNoticeUser(userIdList, noticeId))
                .flatMap(_batch -> redisService.string().del(RedisKeyConst.SEND_NOTICE_USER_ID + loginUser.getId()))
                .as(transactionalOperator::transactional)
                .thenReturn(userIdList));
    }

    private static final String INSERT_NOTICE_SQL = """
            INSERT INTO sys_notice (type, title, content, remark, create_id, create_by, create_time)
            VALUES (:type, :title, :content, :remark, :userId, :userName, NOW())
            """;

    /**
     * 插入通知并获取插入的ID
     *
     * @param notice
     * @param user
     * @return
     */
    private Mono<Long> insertNoticeAndGetId(NoticeDTO notice, LoginUser user) {
        return databaseClient.sql(INSERT_NOTICE_SQL)
                .bind("type", notice.type())
                .bind("title", notice.title())
                .bind("content", notice.content())
                .bind("remark", notice.remark())
                .bind("userId", user.getId())
                .bind("userName", user.getName())
                .filter((statement, executeFunction) -> statement.returnGeneratedValues("id").execute())
                .map(row -> row.get("id", Long.class))
                .one();
    }

    /**
     * 获取未提交的通知
     *
     * @return
     */
    public HandlerFunction<ServerResponse> getNoSubmitNotice() {
        return request -> SecurityUtil.getUserId()
                .flatMap(userId -> redisService.string().get(RedisKeyConst.SEND_NOTICE_USER_ID + userId, NoticeDTO.class))
                .flatMap(ResponseUtil.ok())
                .onErrorResume(ResponseUtil.err());
    }
}
